package net.chengp.kudu_demo;

import java.io.IOException;
import java.security.PrivilegedExceptionAction;
import java.util.ArrayList;
import java.util.List;

import org.apache.hadoop.security.UserGroupInformation;
import org.apache.kudu.ColumnSchema;
import org.apache.kudu.Schema;
import org.apache.kudu.Type;
import org.apache.kudu.client.CreateTableOptions;
import org.apache.kudu.client.Insert;
import org.apache.kudu.client.KuduClient;
import org.apache.kudu.client.KuduClient.KuduClientBuilder;
import org.apache.kudu.client.KuduException;
import org.apache.kudu.client.KuduScanner;
import org.apache.kudu.client.KuduSession;
import org.apache.kudu.client.KuduTable;
import org.apache.kudu.client.ListTablesResponse;
import org.apache.kudu.client.PartialRow;
import org.apache.kudu.client.RowResult;
import org.apache.kudu.client.SessionConfiguration;
import org.junit.Before;
import org.junit.Test;

import jline.internal.Log;

public class TestKudu {
	// 声明全局变量KuduClient后期通过它来操作kudu表
	private KuduClient kuduClient;
	// 指定kuduMaster地址
	private String kuduMaster = "cdh2:7051";
	// 指定表名
	private String tableName = "student";
	// 是否开启kerberos认证
	private boolean enableKerberos = true;

	@Before
	public void init() throws IOException, InterruptedException {
		// 初始化操作
		if(this.enableKerberos) {
			ClientUtils.initKerberosENV(this.enableKerberos);
			kuduClient = UserGroupInformation.getLoginUser().doAs(
					new PrivilegedExceptionAction<KuduClient>() {
						@Override
						public KuduClient run() throws Exception {
							return new KuduClient.KuduClientBuilder(kuduMaster).build();
						}
					}
					);
		}else {
			KuduClientBuilder kuduClientBuilder = new KuduClientBuilder(kuduMaster);
			kuduClient = kuduClientBuilder.build();
		}
	}
	
	@Test
	public void delTable() throws KuduException {
		kuduClient.deleteTable(tableName);
	}

	// 创建表
	@Test
	public void createTable() throws KuduException {
		// 判断表是否存在，不存在就构建
		if (!kuduClient.tableExists(tableName)) {
			// 构建创建表的schema信息 --- 就是表的字段和类型
			ArrayList<ColumnSchema> columnSchemas = new ArrayList<>();
			columnSchemas.add(new ColumnSchema.ColumnSchemaBuilder("id", Type.INT32).key(true).build());
			columnSchemas.add(new ColumnSchema.ColumnSchemaBuilder("name", Type.STRING).build());
			columnSchemas.add(new ColumnSchema.ColumnSchemaBuilder("age", Type.INT32).build());
			columnSchemas.add(new ColumnSchema.ColumnSchemaBuilder("sex", Type.INT32).build());
			Schema schema = new Schema(columnSchemas);
			// 指定创建表的相关属性
			CreateTableOptions options = new CreateTableOptions();
			ArrayList<String> partitionList = new ArrayList<>();
			// 指定kudu表的分区字段是什么
			partitionList.add("id");
			// 按照 id.hashcode % 分区数 = 分区号
			options.addHashPartitions(partitionList, 6);
			kuduClient.createTable(tableName, schema, options);
		}else {
			System.out.println("表已经存在了。。");
		}
	}

	/**
	 * 插入数据
	 */
	@Test
	public void insertTable() throws KuduException {
		// 向表加载数据需要一个kuduSession对象
		KuduSession kuduSession = kuduClient.newSession();
		kuduSession.setFlushMode(SessionConfiguration.FlushMode.AUTO_FLUSH_SYNC);

		// 需要使用kuduTable来构建Operation的子类实例对象
		KuduTable kuduTable = kuduClient.openTable(tableName);

		for (int i = 1; i <= 10; i++) {
			Insert insert = kuduTable.newInsert();
			PartialRow row = insert.getRow();
			row.addInt("id", i);
			row.addString("name", "zhangsan-" + i);
			row.addInt("age", 20 + i);
			row.addInt("sex", i % 2);
			// 最后时限执行数据的加载操作
			kuduSession.apply(insert);
		}
		System.out.println("插入数据完成。");
	}

	@Test
	public void tableInfo() throws KuduException {
		ListTablesResponse response = kuduClient.getTablesList();
		List<String> tables = response.getTablesList();
		System.out.println(tables.toString());
		tables.forEach(e -> {
			try {
				KuduTable table = kuduClient.openTable(e);
				List<ColumnSchema> columns = table.getSchema().getColumns();
				columns.forEach(c -> {
					System.out.println("column=" + c.getName() + ",type=" + c.getType());
				});
			} catch (KuduException e1) {
				e1.printStackTrace();
			}
		});
	}

	@Test
	public void queryRows() throws KuduException {
		KuduTable table = kuduClient.openTable("student");
		KuduScanner scanner = kuduClient.newScannerBuilder(table).build();
		while (scanner.hasMoreRows()) {
			for (RowResult rowResult : scanner.nextRows()) {
				System.out.println(rowResult.getInt("id") + "\t" + rowResult.getString("name"));
			}
		}
	}
	
	@Test
	public void metadata() throws KuduException {
		ListTablesResponse listTablesResponse = kuduClient.getTablesList();
		List<String> tblist = listTablesResponse.getTablesList();
		for (String tableName : tblist) {
			System.out.println(tableName);
		}
	}
}